package com.itcm.wc;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * 描述: 流处理 - 有界
 *
 * @author cm_fighting
 * @create 2022-05-06 上午10:48
 */
public class BoundedStreamWordCount {

    public static void main(String[] args) throws Exception {

        // 1. 创建流式的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 2. 读取文件数据
        DataStreamSource<String> lineDataStreamSource = env.readTextFile("input/words.txt");

        // 3. 数据转换
        SingleOutputStreamOperator<Tuple2<String, Long>> wordAndOneTuple = lineDataStreamSource.flatMap((String line, Collector<Tuple2<String, Long>> out) -> {
            String[] words = line.split(" ");
            for (String word : words) {
                out.collect(Tuple2.of(word, 1L));
            }
        }).returns(Types.TUPLE(Types.STRING, Types.LONG));

        // 4. 分组 指定根据word分组 即元组的f0位置数据
        KeyedStream<Tuple2<String, Long>, String> wordAndOneKeyedStream = wordAndOneTuple.keyBy(data -> data.f0);

        // 5. 按照分组 求和
        SingleOutputStreamOperator<Tuple2<String, Long>> sum = wordAndOneKeyedStream.sum(1);

        // 6. 打印
        sum.print();

        // 7. 启动执行
        /*
            1-6 是定义了运算流程
            7   启动执行
         */
        env.execute();

        /**
         *  5> (world,1)
         *  3> (hello,1)
         *  2> (java,1)
         *  3> (hello,2)
         *  7> (flink,1)
         *  3> (hello,3)
         *
         *  结果分析：
         *      可以看出和批处理的区别，这里数据是一条条处理后，展示当前处理后的结果，不是直接展示处理完后的最终结果，这个容易理解，这也是流处理和批处理逻辑的区别
         *      另外，我们发现输出的处理结果和实际文件中数据的顺序是不一致的，这是因为flink处理数据使用的是大数据引擎，是分布式的架构，当前IDEA环境，其实是用多线程来处理数据的，结果中>前面的数字，指代的是当前本地执行的并行子任务的并行编号
         *      并行子任务个数也是有限制的，取决于当前的并行度，当前IDEA环境下执行且没有设置并行度，默认就是当前执行电脑的CPU核心数
         *      观察可以发现，分组时会把同一个分组的数据分到同一个子任务，这也是必然的 否则不好做统计计算了
         */
    }

}
